#!usr/bin/env python  
# -*- coding:utf-8 _*-
""" 
@author:robot
@file: tpf_dataflow_v2.py 
@version:
@time: 2022/03/14 
"""
from TapUtil.decorator.some_useful import timeout
from TapUtil.logfunc.logger import logger
from TapClientCore.T06_Tapdata_feature.v2.TPF02_dataflows.tpf_dataflow_base import TPFDataFlowBase
from TapException.TM import *
from TapClientCore.T04_logic.DagforDFS.dataflow_body_build.data_copy import DataCopyBuildDFS


class DFSTPFDataFlow(TPFDataFlowBase):
    dataflow_url = None
    DATAFLOW_FUNC = {
        'data copy': DataCopyBuildDFS().data_copy,
    }

    def _create_data_copy(self, data):
        record = {}
        dataflow_type = data.get('dataflow_type')
        dataflow_config = data.get('dataflow_config')
        dataflow_func = self.DATAFLOW_FUNC.get(dataflow_type)
        if dataflow_func is None:
            raise DataFlowBuildError
        dag = dataflow_func(dataflow_config)
        create_response = self.TMF.create_dataflow(dag)
        record['id'] = create_response.get('id')
        record['stages'] = dag
        record['create_response'] = create_response
        return record

    def _delete_dataflow(self, name: str) -> None:
        id_status = self.TMF.query_dataflow_id_status_by_name(name)
        logger.info("DELETE DATAFLOW START")
        if id_status.get('status') in ['paused', 'error']:
            self.TMF.delete_dataflow_by_name(name)
        else:
            self.TMF.manage_dataflow_wait_to_paused_by_name(name)
            self.TMF.delete_dataflow_by_name(name)
        logger.info("DELETE DATAFLOW END")

    @timeout('GET STATS', 20)
    def _get_dataflow_stats(self, name):
        dataflow_stats = self.TMF.query_dataflow_by_name(name).get('stats')
        if dataflow_stats:
            if dataflow_stats.get('input'):
                self.dataflow_stats = dataflow_stats
                return 'GET STATS'

    def create_data_copy(self, data):
        dataflow_name = data.get('dataflow_config').get('config').get('name')
        repeat = self.TMF.dataflow_check_name_repeat(dataflow_name)
        if repeat:
            logger.info(f"{dataflow_name}任务已存在，进行删除")
            dataflow_status = self.TMF.query_dataflow_by_name(dataflow_name).get('status')
            if dataflow_status == 'running':
                self.TMF.manage_stop_and_wait_stop(dataflow_name)
            try:
                self.TMF.manage_delete_dataflow(dataflow_name)
                self.dataflow_delete_mongodb({'name': dataflow_name})
            except Exception as e:
                assert False
        logger.info(f"{dataflow_name}创建任务，并将记录持久化")
        record = self._create_data_copy(data)
        self.dataflow_insert_mongodb(
            {'name': dataflow_name, **record}
        )

    def delete_dataflow(self, name: str) -> None:
        """
        delete dataflow
        :param name:
        :return:
        """
        try:
            self._delete_dataflow(name)
        except DataFlowNotFound as e:
            logger.info(e)
        except DataFlowRemoveFail as e:
            logger.fatal(e)
            assert False
        self.dataflow_delete_mongodb({'name': name})

    def generate_dataflow_url_by_name(self, name):
        """
        http://test.cloud.tapdata.net/console/#/task/622f12223523bc6098eca74b/monitor
        :param name:
        :return:
        """
        dataflow_id = self.TMF.query_dataflow_by_name(name).get('id')
        from TapClientCore.T02_tm_api.v2_api.v2_tm import v2tm as TM
        if TM.tapdata_production == 'dfs':
            self.dataflow_url = f'{TM.serve}/console/#/task/{dataflow_id}/monitor'

    def get_dataflows_status(self, name):
        return self.TMF.query_dataflow_by_name(name).get('status')

    def get_dataflow_stats(self, name):
        self._get_dataflow_stats(name)
        return self.dataflow_stats

    def get_dataflow_stats_insert(self, name):
        return self.get_dataflow_stats(name).get('insert').get('rows')

    def get_dataflow_stats_update(self, name):
        return self.get_dataflow_stats(name).get('update').get('rows')

    def get_dataflow_stats_delete(self, name):
        return self.get_dataflow_stats(name).get('delete').get('rows')

    def dataflow_start_wait_stop(self, dataflow_name):
        self.TMF.manage_start_and_wait_stop(dataflow_name)

    def dataflow_start_wait_running(self, dataflow_name):
        self.TMF.manage_start_and_wait_running(dataflow_name)

    def dataflow_stopping_wait_stop(self, dataflow_name):
        self.TMF.manage_stop_and_wait_stop(dataflow_name)

    def dataflow_is_running(self, dataflow_name):
        dataflow_status = self.TMF.query_dataflow_by_name(dataflow_name).get('status')
        if dataflow_status == 'running':
            return True
        return False
